package cn.doitedu.demo4;

import cn.doitedu.demo4.beans.EventBean;
import cn.doitedu.demo4.beans.RuleMetaBean;
import cn.doitedu.demo4.functions.CdcMeta2RuleMetaBeanFunction;
import cn.doitedu.demo4.functions.RuleProcessCoreFunction;
import cn.doitedu.demo4.utils.EventDataFetcher;
import cn.doitedu.demo4.utils.MetaDataFetcher;
import cn.doitedu.demo4.utils.StateDescUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.nio.ByteBuffer;

public class Demo4 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);


        // 读取规则元数据并转化成javaBean
        SingleOutputStreamOperator<RuleMetaBean> metaBeanStream = MetaDataFetcher.fetchMetaData(env);
        // 广播规则元数据流
        BroadcastStream<RuleMetaBean> metaBeanBroadcastStream = metaBeanStream.broadcast(StateDescUtil.getMetaBcStateDesc());

        // 读取用户行为日志数据
        SingleOutputStreamOperator<EventBean> eventBeanStream = EventDataFetcher.fetchEvents(env);

        // keyBy
        KeyedStream<EventBean, Long> keyedStream = eventBeanStream.keyBy(EventBean::getUser_id);


        // 连接 用户事件流   和   元数据广播流
        SingleOutputStreamOperator<String> messages = keyedStream.connect(metaBeanBroadcastStream)
                .process(new RuleProcessCoreFunction());

        // 将规则命中的结果信息，写入kafka，提供给下游的消息推送引擎去处理
        messages.print();


        env.execute();


    }
}
